package net.alan.etl;

import java.util.Collection;
import java.util.Timer;
import java.util.Vector;

import org.apache.log4j.Logger;

import net.alan.etl.i.IFlowTask;
import net.alan.etl.impl.AbstractFlowTaskTimer;
import net.alan.etl.kafka.FlowParams;
import net.dotool.StringTools;
import net.dotool.reflect.ReflectUtils;

/**
 * @author 李岩飞
 * @email eliyanfei@126.com
 * @date 2018年5月10日下午2:07:17
 */
public class FlowContext {
	static Logger logger = Logger.getLogger(FlowContext.class);
	private static FlowContext flowContext = null;
	private FlowConsumer register;// 初始化注册监听器
	private FlowProductor monitor;//
	private Vector<FlowParams> parmsCache = new Vector<FlowParams>();
	private Timer timer = new Timer();

	private FlowContext() {
		register = new FlowConsumer();
		monitor = new FlowProductor();
		loadTasks();
	}

	public Vector<FlowParams> getParmsCache() {
		return parmsCache;
	}

	/**
	 * 加载执行任务
	 */
	private void loadTasks() {
		ReflectUtils.createSharedReflections("classes", "bin", "flow.");
		Collection<String> subTypes = ReflectUtils.listSubClass(IFlowTask.class);//
		for (final String className : subTypes) {
			try {
				IFlowTask flow = (IFlowTask) Class.forName(className).newInstance();
				String taskId = flow.monitorTaskId();
				if (StringTools.isNotBlank(taskId)) {
					register.subscribe("flow-" + taskId, flow);
				}

				if (flow.timer()) {
					timer((AbstractFlowTaskTimer) flow);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		register.startup();
		monitor.startup();
	}

	public void timer(AbstractFlowTaskTimer task) {
		timer.schedule(task, task.firstDate(), task.period());
		System.out.println("--");
	}

	/**
	 * 实例化上下文类，属于单实例
	 */
	public synchronized static FlowContext createContext() {
		if (flowContext == null) {
			flowContext = new FlowContext();
		}
		return flowContext;
	}

	/**
	 * 通知任务已经完成
	 */
	public void send(FlowParams params) {
		parmsCache.add(params);
	}

	public void print(String message) {
		logger.info(message);
	}
}
